/*************************************************************************************
* 	 Copyright (C) 2010 by Information Systems Group, Saarland University  			*
*    http://infosys.cs.uni-saarland.de												*
* 	 																				*
* 	 This file is part of Hadoop++.												 	*
*																					*
*    Hadoop++ is free software: you can redistribute it and/or modify				*
*    it under the terms of the GNU Lesser General Public License as published by	*
*    the Free Software Foundation, either version 3 of the License, or				*
*    (at your option) any later version.											*
*																					*
*    Hadoop++ is distributed in the hope that it will be useful,					*
*    but WITHOUT ANY WARRANTY; without even the implied warranty of					*
*    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the					*
*    GNU Lesser General Public License for more details.							*
*																					*
*    You should have received a copy of the GNU Lesser General Public License		*
*    along with Hadoop++.  If not, see <http://www.gnu.org/licenses/>.				*
*************************************************************************************/
package unisb.cs.core.index.access;

import java.io.IOException;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;

import unisb.cs.data.tables.Record;
import unisb.cs.data.tables.TableObjectProxy;

/**
 * This class process the split data using index access and passes
 * records to the mapper.
 *
 */
public class IndexAccesserRecordReader implements RecordReader<Text, Text> {
	// split variables
	protected FileSplit fileSplit;
	protected FSDataInputStream in;
	protected int splitSize = 0;
	private int offset = 0;
	protected byte[] contents;
	
	private int[] valueFieldIndex;
	private String tableName;
	// index variables
	private int key, keyType, keySize;
	
	// record iterator 
	private IndexLookup.RecordList records = null;

	/**
	 * Prepare the record reader.
	 * This method performs an index lookup 
	 * and sets the record iterator to the 
	 * first record in referenced portion of 
	 * the split.
	 * @throws Exception 
	 */
	public IndexAccesserRecordReader(FileSplit fileSplit, JobConf job) throws Exception {
		this.fileSplit = fileSplit;
		Path file = fileSplit.getPath();
		FileSystem fs = file.getFileSystem(job);
		this.in = fs.open(file);
		splitSize = (int) fileSplit.getLength();
		IOUtils.skipFully(in, fileSplit.getStart());
		
		if(job.get("value.field.index") != null){
			String [] valueStringFieldIndex = job.get("value.field.index").split("|");
			valueFieldIndex = new int[valueStringFieldIndex.length];
			for(int i = 0; i < valueStringFieldIndex.length; i++)
				valueFieldIndex[i] = Integer.parseInt(valueStringFieldIndex[i]);
		}
		else
			valueFieldIndex = null;
		
		tableName = job.get("table.name");
			
		// set index variables
		key = Integer.parseInt(job.get(IndexAccesserInputFormat.IDX_ATTR));
		keyType = Integer.parseInt(job.get(IndexAccesserInputFormat.IDX_ATTR_TYPE));
		keySize = Integer.parseInt(job.get(IndexAccesserInputFormat.IDX_ATTR_SIZE));
		
		/**
		 * Perform index lookup.
		 * Internally, it first reads the index header
		 * to check whether or not the split contains any
		 * referenced data or not. Only in case it does that
		 * the index is loaded and looked up.   
		 */
		IndexLookup e = new IndexLookup(key, keyType, keySize, tableName);
		if (fileSplit.getLength() > 0)
			records = e.extract(in, job
					.get(IndexAccesserInputFormat.LOW_KEY), job
					.get(IndexAccesserInputFormat.HIGH_KEY), fileSplit
					.getStart(), fileSplit.getStart() + fileSplit.getLength()
					- 1);
	}

	public Text createKey() {
		return new Text();
	}

	public Text createValue() {
		return new Text();
	}

	public long getPos() throws IOException {
		return fileSplit.getStart() + offset;
	}

	public float getProgress() throws IOException {
		return (float)(in.getPos() - fileSplit.getStart()) / splitSize;
	}

	/**
	 * We override the next method so as to get all the the <K, V> per line of
	 * the InputSplit.
	 */
	public boolean next(Text key, Text value) throws IOException {
		// iterate over the relevant records
		if (records != null && records.hasNext()) {
			Record r = records.next();
			try{
				TableObjectProxy proxy = new TableObjectProxy(tableName, r);
				proxy.getByteAttributes(key, this.key);
				proxy.getStringAttributes(value, valueFieldIndex);
			} catch (Exception e) {
				e.printStackTrace();
			}
			return true;
		} else
			return false;
	}

	public void close() throws IOException {
		IOUtils.closeStream(in);
	}
}